package com.atguigu.flinkState;


import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author wky
 * @create 2021-07-19-16:54
 */

//广播 变量 又一个流决定另一流的执行状态
public class State_Broadcast {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        DataStreamSource<String> source1 = env.socketTextStream("hadoop102", 9999);
        DataStreamSource<String> source2 = env.socketTextStream("hadoop103", 9999);

        //广播变量配置
        MapStateDescriptor<String, String> myBroadcast = new MapStateDescriptor<>("broadcast", String.class, String.class);
        //广播
        BroadcastStream<String> broadcast = source2.broadcast(myBroadcast);

        source1
                .connect(broadcast)
                .process(new BroadcastProcessFunction<String, String, String>() {
            @Override
            public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                ReadOnlyBroadcastState<String, String> state = ctx.getBroadcastState(myBroadcast);
                if ("1".equals(state.get("switch"))){
                    out.collect("1号配置");
                }
                else if("2".equals(state.get("switch"))){
                    out.collect("二号配置");
                }
                else {
                    out.collect("三号配置 ");
                }
            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(myBroadcast);
                broadcastState.put("switch",value);

            }
        }).print();
        env.execute();
    }
}
